
package com.lg.monitor

import java.sql.{Connection, DriverManager, Statement}

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter

object MySQLWriter {

  //获取mysql连接
  def getConnection: Connection ={

    Class.forName("com.mysql.jdbc.Driver")
    val connection: Connection = DriverManager
      .getConnection("jdbc:mysql://server3:3306/lg_metrics?user=root&password=123456&useSSL=false")
    connection
  }

  class MySQLWriter extends ForeachWriter[BusInfo]{

    var conn: Connection = _
    var statement: Statement =_
    override def open(partitionId: Long, epochId: Long): Boolean = {

      conn = MySQLWriter.getConnection
      statement = conn.createStatement()
      true
    }

    override def process(value: BusInfo): Unit = {
      //向mysql写入数据
      val lglat: String = value.lglat
      val deployNum: String = value.deployNum
      val sql = "insert into lg_metrics.lglat_info values('"+deployNum+"','"+lglat+"')"
      statement.executeUpdate(sql)
    }

    //关闭连接
    override def close(errorOrNull: Throwable): Unit = {
      conn.close()
    }
  }


}

